package com.xuelang.pipeline.app.log.handler;

import com.xuelang.pipeline.app.log.client.LogWebSocketClient;
import com.xuelang.pipeline.client.common.constant.Constant;
import com.xuelang.pipeline.client.common.util.BizAssert;
import com.xuelang.pipeline.domain.resource.Cluster;
import com.xuelang.pipeline.domain.resource.Environment;
import com.xuelang.pipeline.domain.resource.Space;
import com.xuelang.pipeline.domain.resource.service.ClusterDomainService;
import com.xuelang.pipeline.domain.resource.service.EnvironmentDomainService;
import com.xuelang.pipeline.domain.resource.service.SpaceDomainService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import org.springframework.web.util.UriComponentsBuilder;

import jakarta.annotation.Resource;
import java.io.IOException;
import java.net.URI;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * <p>日志webSocket处理器</p>
 *
 * @author : yixin.lys
 * @date : 2024-07-01 10:06
 **/
@Slf4j
public class LogWebSocketHandler extends TextWebSocketHandler {

    private ScheduledExecutorService heartbeatExecutor;

    private LogWebSocketClient client;

    @Resource
    private EnvironmentDomainService environmentDomainService;

    @Resource
    private SpaceDomainService spaceDomainService;

    @Autowired
    private ClusterDomainService clusterDomainService;

    @Value("${pipeline.worker.url:}")
    private String workerAddress;

    private final String TASK_LOG_PATH_TRAIT = "taskLog";

    /**
     * 错过心跳次数
     */
    private int missedHeartbeats = 0;
    /**
     * 允许的最大错过次数
     */
    private static final int MAX_MISSED_HEARTBEATS = 3;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        super.afterConnectionEstablished(session);

        // 启动心跳机制
        heartbeatExecutor = Executors.newScheduledThreadPool(1);
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                if (session.isOpen()) {
                    session.sendMessage(new TextMessage("ping"));
                    log.info("Sent ping to client");
                    missedHeartbeats = 0; // Reset missed heartbeats on successful ping
                }
                missedHeartbeats++;

                if (missedHeartbeats > MAX_MISSED_HEARTBEATS) {
                    log.warn("Missed too many heartbeats, closing session");
                    session.close(CloseStatus.GOING_AWAY);
                    shutdownHeartbeatExecutor();
                }
            } catch (IOException e) {
                log.error("Error sending ping to client: " + e.getMessage(), e);
                shutdownHeartbeatExecutor();
            }
        }, 0, 30, TimeUnit.SECONDS);

        // 获取客户端传递的参数
        URI uri = session.getUri();
        String path = uri.getPath();
        String query = uri.getQuery();
        String envId = "";
        String podName = "";
        String containerName = "";
        int sinceSeconds = 0;
        int tailLines = 500;

        if (query != null) {
            String[] params = query.split("&");
            for (String param : params) {
                String[] keyValue = param.split("=");
                if (keyValue.length == 2) {
                    switch (keyValue[0]) {
                        case "envId":
                            envId = keyValue[1];
                            break;
                        case "podName":
                            podName = keyValue[1];
                            break;
                        case "containerName":
                            containerName = keyValue[1];
                            break;
                        case "sinceSeconds":
                            sinceSeconds = Integer.parseInt(keyValue[1]);
                            break;
                        case "tailLines":
                            tailLines = Integer.parseInt(keyValue[1]);
                            break;
                    }
                }
            }
        }

        // 根据环境id查询KubeConfig和namespace
        Environment environment = environmentDomainService.query(envId);
        BizAssert.notNull(environment, "环境不存在");
        Space space = spaceDomainService.query(environment.getSpaceUuid());
        BizAssert.notNull(space, "交付空间不存在");
        Cluster cluster = clusterDomainService.query(space.getClusterUuid());
        BizAssert.notNull(cluster, "集群不存在");

        // 根据路由来决定实际的namespace，若是任务日志，则统一使用默认namespace
        String namespace;
        if (path.contains(TASK_LOG_PATH_TRAIT)) {
            namespace = Constant.BUILD_NAMESPACE;
        }else{
            namespace = space.getNamespace();
        }

        URI logSocketUri = UriComponentsBuilder.fromUriString("ws://" + workerAddress + "/log-provider-server-url")
                .queryParam("namespace", namespace)
                .queryParam("podName", podName)
                .queryParam("containerName", containerName)
                .queryParam("sinceSeconds", sinceSeconds)
                .queryParam("tailLines", tailLines)
                .queryParam("kubeConfig", cluster.getKubeConfig())
                .build().toUri();
        client = new LogWebSocketClient(logSocketUri, session, heartbeatExecutor);
        client.connect();
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
        shutdownHeartbeatExecutor();
        // 清理资源，关闭 WebSocket 客户端
        if (Objects.nonNull(client) && client.isOpen()) {
            client.close();
        }
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // 处理来自前端的消息
        String payload = message.getPayload();
        if ("pong".equals(payload)) {
            log.info("Received pong from client");
            missedHeartbeats = 0; // Reset missed heartbeats on receiving pong
        } else {
            // 处理其他消息
            log.info("Received message: {}", payload);
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
        shutdownHeartbeatExecutor();
    }

    private void shutdownHeartbeatExecutor() {
        if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) {
            heartbeatExecutor.shutdownNow();
        }
    }

}
